/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server.persistence;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.zip.CheckedInputStream;
import java.util.zip.CheckedOutputStream;

import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.jute.InputArchive;
import org.apache.jute.OutputArchive;
import org.apache.zookeeper.server.DataTree;
import org.apache.zookeeper.server.util.SerializeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**实现Snapshot接口，负责存储、序列化、反序列化、访问快照
 * <br/>This class implements the snapshot interface.
 * <br/>it is responsible for storing, serializing and deserializing the right snapshot.
 * <br/>and provides access to the snapshots.
 */
public class FileSnap implements SnapShot {

    /**snapshot目录文件
     *
     */
    File snapDir;
    SnapshotInfo lastSnapshotInfo = null;
    /**是否已经关闭标识
     *
     */
    private volatile boolean close = false;
    /**版本号
     *
     */
    private static final int VERSION = 2;
    /**database id
     *
     */
    private static final long dbId = -1;
    private static final Logger LOG = LoggerFactory.getLogger(FileSnap.class);
    /**snapshot文件的魔数(类似class文件的魔数)
     *
     */
    public static final int SNAP_MAGIC = ByteBuffer.wrap("ZKSN".getBytes()).getInt();

    public static final String SNAPSHOT_FILE_PREFIX = "snapshot";

    public FileSnap(File snapDir) {
        this.snapDir = snapDir;
    }

    /**
     * get information of the last saved/restored snapshot
     * @return info of last snapshot
     */
    @Override
	public SnapshotInfo getLastSnapshotInfo() {
        return this.lastSnapshotInfo;
    }

    /**deserialize主要用作反序列化，并将反序列化结果保存至dt和sessions中。 其大致步骤如下:
     * <br/>1、获取100个合法的snapshot文件，并且snapshot文件已经通过zxid进行降序排序，进入2
     * <br/>2、遍历100个snapshot文件，从zxid最大的开始，读取该文件，并创建相应的InputArchive，进入3
     * <br/>3、调用deserialize(dt, sessions, ia)函数完成反序列化操作，进入4
     * <br/>4、验证从文件中读取的Checksum是否与新生的Checksum相等，若不等，则抛出异常，否则，进入5
     * <br/>5、跳出循环并关闭相应的输入流，并从文件名中解析出相应的zxid返回。
     * <br/>6、在遍历100个snapshot文件后仍然无法找到通过验证的文件，则抛出异常。
     * <br/>deserialize a data tree from the most recent snapshot
     * @return the zxid of the snapshot
     */
    @Override
	public long deserialize(DataTree dt, Map<Long, Integer> sessions) throws IOException {
        // we run through 100 snapshots (not all of them)
        // if we cannot get it running within 100 snapshots we should  give up
    	// 查找100个合法的snapshot文件
        List<File> snapList = findNValidSnapshots(100);
        if (snapList.size() == 0) { // 无snapshot文件，直接返回
            return -1L;
        }
        File snap = null;
        long snapZxid = -1;
        // 默认为不合法
        boolean foundValid = false;
        for (int i = 0, snapListSize = snapList.size(); i < snapListSize; i++) { // 遍历snapList
            snap = snapList.get(i);
            LOG.info("Reading snapshot {}", snap);
            // 从文件名中解析出zxid
            snapZxid = Util.getZxidFromName(snap.getName(), SNAPSHOT_FILE_PREFIX);
            // 读取指定的snapshot文件
            try (CheckedInputStream snapIS = SnapStream.getInputStream(snap)) {
                InputArchive ia = BinaryInputArchive.getArchive(snapIS);
                // 反序列化
                deserialize(dt, sessions, ia);
                SnapStream.checkSealIntegrity(snapIS, ia);

                // Digest feature was added after the CRC to make it backward
                // compatible, the older code can still read snapshots which
                // includes digest.
                //
                // To check the intact, after adding digest we added another
                // CRC check.
                if (dt.deserializeZxidDigest(ia, snapZxid)) {
                    SnapStream.checkSealIntegrity(snapIS, ia);
                }

                // 合法
                foundValid = true;
                // 从最新的文件开始找，只要有一个合法就跳出循环
                break;
            } catch (IOException e) {
                LOG.warn("problem reading snap file {}", snap, e);
            }
        }
        if (!foundValid) { // 遍历所有文件都未验证成功
            throw new IOException("Not able to find valid snapshots in " + snapDir);
        }
        // lastProcessedZxid: snapshot里最新文件名解析出的zxid
        dt.lastProcessedZxid = snapZxid;
        lastSnapshotInfo = new SnapshotInfo(dt.lastProcessedZxid, snap.lastModified() / 1000);

        // compare the digest if this is not a fuzzy snapshot, we want to compare
        // and find inconsistent asap.
        // 如果这不是一个模糊快照就比较摘要，我们想要比较和尽快发现不一致
        if (dt.getDigestFromLoadedSnapshot() != null) {
            dt.compareSnapshotDigests(dt.lastProcessedZxid);
        }
        return dt.lastProcessedZxid;
    }

    /**反序列化，并将反序列化结果保存至header和sessions中。其中会验证header的魔数是否相等
     * <br/>deserialize the datatree from an inputarchive
     * @param dt the datatree to be serialized into
     * @param sessions the sessions to be filled up
     * @param ia the input archive to restore from
     * @throws IOException
     */
    public void deserialize(DataTree dt, Map<Long, Integer> sessions, InputArchive ia) throws IOException {
        FileHeader header = new FileHeader();
        // 反序列化至header
        header.deserialize(ia, "fileheader");
        if (header.getMagic() != SNAP_MAGIC) { // 验证魔数是否相等
            throw new IOException("mismatching magic headers " + header.getMagic() + " !=  " + FileSnap.SNAP_MAGIC);
        }
        // 反序列化至dt、sessions
        SerializeUtils.deserializeSnapshot(dt, ia, sessions);
    }

    /**
     * find the most recent snapshot in the database.
     * @return the file containing the most recent snapshot
     */
    @Override
	public File findMostRecentSnapshot() throws IOException {
        List<File> files = findNValidSnapshots(1);
        if (files.size() == 0) {
            return null;
        }
        return files.get(0);
    }

    /**先把snapshot文件按序排序，然后获取符合规则的最新的n（100）个文件返回。SnapStream的isValidSnapshot函数主要是从文件名和文件的结尾符号是否是"/"来判断snapshot文件是否合法
     * <br/>find the last (maybe) valid n snapshots. this does some
     * minor checks on the validity of the snapshots. It just
     * checks for / at the end of the snapshot. This does
     * not mean that the snapshot is truly valid but is
     * valid with a high probability. also, the most recent
     * will be first on the list.
     * @param n the number of most recent snapshots
     * @return the last n snapshots (the number might be
     * less than n in case enough snapshots are not available).
     * @throws IOException
     */
    protected List<File> findNValidSnapshots(int n) throws IOException {
    	// 按照zxid对snapshot文件进行降序排序
        List<File> files = Util.sortDataDir(snapDir.listFiles(), SNAPSHOT_FILE_PREFIX, false);
        int count = 0;
        List<File> list = new ArrayList<File>();
        for (File f : files) { // 遍历snapshot文件
            // we should catch the exceptions from the valid snapshot and continue until we find a valid one
        	// 我们应该从有效快照捕获异常，并继续执行，直到找到一个有效快照为止
            try {
            	// 验证文件是否合法，在写snapshot文件时服务器宕机，此时的snapshot文件非法；非snapshot文件也非法
                if (SnapStream.isValidSnapshot(f)) {
                	// 合法则添加
                    list.add(f);
                    // 计数器加一
                    count++;
                    if (count == n) { // 等于n则跳出循环
                        break;
                    }
                }
            } catch (IOException e) {
                LOG.warn("invalid snapshot {}", f, e);
            }
        }
        return list;
    }

    /**
     * find the last n snapshots. this does not have
     * any checks if the snapshot might be valid or not
     * @param n the number of most recent snapshots
     * @return the last n snapshots
     * @throws IOException
     */
    public List<File> findNRecentSnapshots(int n) throws IOException {
        List<File> files = Util.sortDataDir(snapDir.listFiles(), SNAPSHOT_FILE_PREFIX, false);
        int count = 0;
        List<File> list = new ArrayList<File>();
        for (File f : files) {
            if (count == n) {
                break;
            }
            if (Util.getZxidFromName(f.getName(), SNAPSHOT_FILE_PREFIX) != -1) {
                count++;
                list.add(f);
            }
        }
        return list;
    }

    /**用于序列化dt、sessions和header，其中，首先会检查header是否为空，然后依次序列化header，sessions和dt
     * <br/>serialize the datatree and sessions
     * @param dt the datatree to be serialized
     * @param sessions the sessions to be serialized
     * @param oa the output archive to serialize into
     * @param header the header of this snapshot
     * @throws IOException
     */
    protected void serialize(DataTree dt, Map<Long, Integer> sessions, OutputArchive oa, FileHeader header) throws IOException {
        // this is really a programmatic error and not something that can
        // happen at runtime
    	// 文件头为null
        if (header == null) {
            throw new IllegalStateException("Snapshot's not open for writing: uninitialized header");
        }
        // 将header序列化
        header.serialize(oa, "fileheader");
        // 将dt、sessions序列化
        SerializeUtils.serializeSnapshot(dt, oa, sessions);
    }

    /**该函数用于将header、sessions、dt序列化至本地snapshot文件中，并且在最后会写入"/"字符。该方法是同步的，即是线程安全的
     * <br/>serialize the datatree and session into the file snapshot
     * @param dt the datatree to be serialized
     * @param sessions the sessions to be serialized
     * @param snapShot the file to store snapshot into
     * @param fsync sync the file immediately after write
     */
    @Override
	public synchronized void serialize(DataTree dt, Map<Long, Integer> sessions, File snapShot, boolean fsync) throws IOException {
        if (!close) { // 未关闭
        	// 输出流
            try (CheckedOutputStream snapOS = SnapStream.getOutputStream(snapShot, fsync)) {
                OutputArchive oa = BinaryOutputArchive.getArchive(snapOS);
                // 新生文件头
                FileHeader header = new FileHeader(SNAP_MAGIC, VERSION, dbId);
                // 序列化dt、sessions、header
                serialize(dt, sessions, oa, header);
                SnapStream.sealStream(snapOS, oa);

                // Digest feature was added after the CRC to make it backward
                // compatible, the older code cal still read snapshots which
                // includes digest.
                //
                // To check the intact, after adding digest we added another
                // CRC check.
                if (dt.serializeZxidDigest(oa)) {
                    SnapStream.sealStream(snapOS, oa);
                }

                lastSnapshotInfo = new SnapshotInfo(
                    Util.getZxidFromName(snapShot.getName(), SNAPSHOT_FILE_PREFIX),
                    snapShot.lastModified() / 1000);
            }
        } else {
            throw new IOException("FileSnap has already been closed");
        }
    }

    /**
     * synchronized close just so that if serialize is in place
     * the close operation will block and will wait till serialize
     * is done and will set the close flag
     */
    @Override
    public synchronized void close() throws IOException {
        close = true;
    }

}
